#include "cgzCTcpAsioC.h"
#include "cgzCThreadPool.h"

cgzCTcpAsioC::cgzCTcpAsioC(
    cgzCThreadPool *pool,
    const char *name,
    const std::string ip,
    const int port,
    const int cid)
    : cgzCTcpC(pool, name, ip, port, cid),
      m_io_context(nullptr),
      m_socket(nullptr)
{
}

cgzCTcpAsioC::~cgzCTcpAsioC()
{
    m_io_context->stop();
    m_socket->close();
    delete m_socket;
    delete m_io_context;
}

bool cgzCTcpAsioC::tcpinit()
{
    m_io_context = new io_context();
    m_socket = new tcp::socket(*m_io_context);

    tcp::resolver resolver(*m_io_context);
    boost::system::error_code ec;
    std::string port_str = std::to_string(m_port);
    m_endpoints = resolver.resolve(m_ip, port_str.c_str(), ec);
    if (ec)
    {
        cgz_this_error("resolve error: {}", ec.message().c_str());
        return false;
    }
    return tcplink();
}
bool cgzCTcpAsioC::tcplink()
{
    try
    {
        boost::asio::connect(*m_socket, m_endpoints);
        m_fd = m_socket->native_handle();
        m_hand = true;
        sendaccept();

        // 注册服务
        service();
        // 创建事件线程
        m_thread = new std::thread(&cgzCTcpAsioC::do_tgoing, this);
        m_pool->push_thread(m_thread);
        return true;
    }
    catch (std::exception &e)
    {
        cgz_this_error("{}", e.what());
    }
    return false;
}

bool cgzCTcpAsioC::run(int64_t nms)
{
    while (!m_runTask.empty())
    {
        auto task = m_runTask.front();
        m_runTask.pop_front();
        auto loop = static_cast<cgzCLoopC *>(task);
        if (loop->type == TCPC_SEND)
        {
            auto csend = static_cast<cgzCSendC *>(loop);
            async_write(*m_socket, buffer(csend->m_data, csend->m_len),
                        [this, csend](boost::system::error_code ec, std::size_t)
                        {
                            if (ec)
                            {
                                std::unique_lock lock(m_rwmutex);
                                disctcp();
                                cgz_this_error("err:{}", ec.message().c_str());
                            }
                            csend->~cgzCSendC();
                            free(csend);
                        });
        }
        else
        {
            cgzCTcpC::run(loop);
        }
    }

    return true;
}

void cgzCTcpAsioC::do_tgoing()
{
    tcpread();
    while (m_bRunning)
    {
        m_io_context->run();
    }
}

bool cgzCTcpAsioC::tcpread()
{
    m_socket->async_read_some(buffer(m_readCire.head(), m_readCire.dire()),
                              [this](boost::system::error_code ec, size_t tlen)
                              {
                                  if (!ec)
                                  {
                                      std::unique_lock lock(m_rwmutex);
                                      m_readCire.push(tlen);
                                      while (sticking())
                                          ;
                                      tcpread();
                                  }
                                  else
                                  {
                                      cgz_this_error("err:{}", ec.message().c_str());
                                  }
                              });
    return true;
}

bool cgzCTcpAsioC::tcpsend()
{
    assert(false);
    return false;
}
